#include "my_disk_buf.h"
void
my_disk_buf_clean(my_disk_buf_t *buf) {
    if (my_atomic_int_load(&buf->read_counter) != 0) {
        MY_LOG_FATAL0("disk_buf", "the current buffer has readers and "
            "cannot be clean it");
    }

    my_spin_lock_lock(&buf->lock);
    buf->write_serial = 0;
    buf->start_off = 0;
    buf->write_off = 0;
    my_spin_lock_unlock(&buf->lock);
}

my_cache_err_t    // for async write
my_disk_buf_write_hold(my_disk_buf_t *buf, int64_t aligned_size,
    int64_t *offset_in_buffer) {

    if (!MY_IS_PAGE_ALIGNED(aligned_size) || aligned_size > MY_DISK_BLOCK_SIZE) {
        MY_LOG_FATAL0("disk_buf", "invalid aligned_size");
    }

    my_cache_err_t err = CACHE_ERR_NO_SPACE;
    my_spin_lock_lock(&buf->lock);
    if (buf->write_off + aligned_size <= MY_DISK_BLOCK_SIZE) {
        *offset_in_buffer = buf->write_off;
        my_atomic_int_increment_and_fetch(&buf->write_counter, 1);
        err = CACHE_ERR_NO_ERR;
    }
    my_spin_lock_unlock(&buf->lock);

    return err;
}

void // async write
my_disk_buf_write(my_disk_buf_t *buf, my_cache_key_t *key, my_buf_chain_t *data,
    uint32_t crc32, int64_t offset_in_buffer) {

    my_disk_cache_object_t obj;
    obj.key_len = key->key_len;
    obj.data_len = my_buf_chain_total_bytes(data);
    obj.crc32 = crc32;

    int64_t object_size = sizeof(my_disk_cache_object_t) + obj.key_len + obj.data_len;
    int64_t aligned_size = MY_ALIGN_PAGE_SIZE(object_size);
    int64_t pending_size = aligned_size - object_size;

    int64_t cur_off = offset_in_buffer;
    memcpy(buf->buf + cur_off, &obj, sizeof(my_disk_cache_object_t));
    cur_off += sizeof(my_disk_cache_object_t);
    memcpy(buf->buf + cur_off, key->key, obj.key_len);
    cur_off += obj.key_len;

    int64_t copy_len = 0;
    my_buf_t *mbuf = NULL;
    while((mbuf = my_buf_chain_subtract(data)) != NULL) {
        int64_t data_len = mbuf->end - mbuf->pos;
        copy_len += data_len;
        if (copy_len > obj.data_len) {
            MY_LOG_FATAL0("disk_buf", "invalid data len");
        }
        memcpy(buf->buf + cur_off, mbuf->pos, data_len);
        cur_off += data_len;
        my_buf_free(mbuf);
    }

    if (copy_len != obj.data_len) {
        MY_LOG_FATAL0("disk_buf", "invalid data len");
    }

    bzero(buf->buf + cur_off, pending_size);

    my_atomic_int_increment_and_fetch(&buf->write_counter, -1);
}

static my_cache_err_t
my_disk_buf_read_hold_helper(my_disk_buf_t *buf, uint8_t write_serial,
    int64_t offset, int64_t size) {

    if (buf->write_serial != write_serial || offset < buf->start_off ||
        offset > buf->start_off + MY_DISK_BLOCK_SIZE) {
        return CACHE_ERR_ALREADY_FLUSHED;
    }

    if (size < sizeof(my_disk_cache_object_t)) {
        MY_LOG_ERROR0("disk_buf", "invalid size argment");
    }

    int64_t offset_in_buffer = offset - buf->start_off;
    if (offset_in_buffer + size > MY_DISK_BLOCK_SIZE) {
        MY_LOG_ERROR0("disk_buf", "invalid offset and size argments in current buffer");
        return CACHE_ERR_INVALID;
    }

    my_atomic_int_increment_and_fetch(&buf->read_counter, 1);
    return CACHE_ERR_NO_ERR;
}

my_cache_err_t // for async read
my_disk_buf_read_hold(my_disk_buf_t *buf, uint8_t write_serial,
    int64_t offset, int64_t size) {

    if (!MY_IS_PAGE_ALIGNED(offset) || !MY_IS_PAGE_ALIGNED(size)) {
        MY_LOG_FATAL0("disk_buf", "params error");
    }

    my_cache_err_t err;
    my_spin_lock_lock(&buf->lock);
    err = my_disk_buf_read_hold_helper(buf, write_serial, offset, size);
    my_spin_lock_unlock(&buf->lock);
    return err;
}

my_cache_err_t
my_disk_buf_read_helper(my_disk_buf_t *buf, my_cache_key_t *key, int64_t offset,
    int64_t size, my_buf_chain_t *data, my_buf_t *mbuf) {

    int64_t offset_in_buffer = offset - buf->start_off;
    memcpy(mbuf->end, buf->buf + offset_in_buffer, size);
    mbuf->end += size;

    my_disk_cache_object_t *obj = (my_disk_cache_object_t *)mbuf->pos;
    if (key->key_len != obj->key_len ||
        memcmp(key->key, mbuf->pos + sizeof(my_disk_cache_object_t), key->key_len) != 0) {
        MY_LOG_ERROR0("disk_buf", "key no matched in disk buf");
        return CACHE_ERR_INVALID;
    }

    if (size < sizeof(my_disk_cache_object_t) + obj->key_len + obj->data_len) {
        MY_LOG_ERROR0("disk_buf", "to short object");
        return CACHE_ERR_INVALID;
    }

    // adjust mbufer pos and end
    mbuf->pos += sizeof(my_disk_cache_object_t) + obj->key_len;
    mbuf->end = mbuf->pos + obj->data_len;

    uint32_t crc = my_buf_crc32(mbuf);
    if (crc != obj->crc32) {
        MY_LOG_ERROR0("disk_buf", "crc32 no mathced");
        return CACHE_ERR_INVALID;
    }

    my_buf_chain_append(data, mbuf);
    return CACHE_ERR_NO_ERR;
}

my_cache_err_t   // async read
my_disk_buf_read(my_disk_buf_t *buf, my_cache_key_t *key, int64_t offset,
    int64_t size, my_buf_chain_t *data) {

    my_buf_data_t *buf_data = my_buf_data_alloc(size);
    my_buf_t *mbuf = my_buf_alloc(buf_data, buf_data->data, buf_data->data);
    my_buf_data_free(buf_data); // free buf_data
    my_cache_err_t err = my_disk_buf_read_helper(buf, key, offset,
        size, data, mbuf);
    my_buf_free(mbuf);  // free mbuf
    my_atomic_int_increment_and_fetch(&buf->read_counter, -1);
    return err;
}

my_disk_buf_t *
my_disk_buf_pool_alloc_helper(my_disk_buf_pool_t *pool, uint8_t write_serial,
    int64_t start_offset) {

    if (pool->cur_bufs <= 0) {
        return NULL;
    }

    my_disk_buf_t *buf = NULL;
    my_queue_t *q = my_queue_last(&pool->head);
    buf = my_queue_data(q, my_disk_buf_t, queue);
    if (!my_disk_buf_can_clean(buf)) {
        return NULL; // can't clean, please try again
    }

    my_queue_remove(q);
    pool->cur_bufs--;

    my_disk_buf_clean(buf);
    buf->write_serial = write_serial;
    buf->start_off = start_offset;

    return buf;
}

my_disk_buf_t *
my_disk_buf_pool_alloc(my_disk_buf_pool_t *pool, uint8_t write_serial,
    int64_t start_offset) {

    my_disk_buf_t *buf = NULL;
    my_spin_lock_lock(&pool->lock);
    buf = my_disk_buf_pool_alloc_helper(pool, write_serial, start_offset);
    my_spin_lock_unlock(&pool->lock);

    return buf;
}

void
my_disk_buf_pool_free(my_disk_buf_pool_t *pool, my_disk_buf_t *buf) {
    my_spin_lock_lock(&pool->lock);
    my_queue_insert_head(&pool->head, &buf->queue);
    pool->cur_bufs++;
    my_spin_lock_unlock(&pool->lock);
}